package com.laicunba.service.mq;

import java.net.InetAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory; 
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.laicunba.constant.GloabConstant;
import com.laicunba.pojo.MessageDO;
import com.laicunba.service.MessageService;
import com.laicunba.util.StringUtil; 

@Service
public class ActiveMQServiceImpl implements ActiveMQService {

	private static final ExecutorService executor = Executors.newFixedThreadPool(1);
	private static final Logger LOG = Logger.getLogger(ActiveMQServiceImpl.class);
	private static final String USER = GloabConstant.ACTIVEMQ_USERNAME;
	private static final String PASSWORD = GloabConstant.ACTIVEMQ_PASSWORD;
	private static final String URL = GloabConstant.ACTIVEMQ_URL;
	private static ActiveMQConnectionFactory FACTORY;
	private static Connection CONNECTION;
	private static boolean MOCK = StringUtil.isEmpty(URL);

	private static long TOPIC_TIME_TO_LIVE = 1000 * 60 * 60 * 24;
	private static Map<String,Session> CONSUMMER_MAP = new HashMap();

	@Autowired
	private MessageService messageService;

	private static void initconn() {
		if (CONNECTION != null) {
			return;
		}
		executor.execute(new Runnable() {

			@Override
			public void run() {
				try {
					MOCK = true;
					FACTORY = new ActiveMQConnectionFactory(URL);
					CONNECTION = FACTORY.createConnection(USER, PASSWORD);
					if (StringUtil.isNotEmpty(GloabConstant.ACTIVEMQ_CLIENTID)) {
						CONNECTION.setClientID(GloabConstant.ACTIVEMQ_CLIENTID + ".sender." + InetAddress.getLocalHost().getHostName() + "." + System.currentTimeMillis());
						 
					}
					CONNECTION.start();
					MOCK = false;
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		});

	}

	static {
		initconn();
	}

	@Override
	public boolean sendTopics(final String topics, final String msgStr) {
		if (MOCK) {
			return false;
		}
		executor.execute(new Runnable() {

			@Override
			public void run() {
				try {

					Session session = CONNECTION.createSession(false, Session.AUTO_ACKNOWLEDGE);
					Destination dest = new ActiveMQTopic(topics);
					MessageProducer producer = session.createProducer(dest);
					producer.setDeliveryMode(DeliveryMode.PERSISTENT);
					TextMessage msg = session.createTextMessage(msgStr);
					producer.setTimeToLive(TOPIC_TIME_TO_LIVE);
					producer.send(msg);
					session.close();

				} catch (JMSException e) {
					LOG.error("SEND_MQ_ERROR", e);
				}

			}
		});

		return true;
	}

	@Override
	public boolean sendQueues(final String name, final String msgStr, final String tid) {
		if (MOCK) {
			return false;
		}

		executor.execute(new Runnable() {

			@Override
			public void run() {
				try {

					// Session session = CONNECTION.createSession(false,
					// Session.AUTO_ACKNOWLEDGE);
					Session session = CONNECTION.createSession(true, Session.SESSION_TRANSACTED);
					Destination dest = new ActiveMQQueue(name);

					MessageProducer producer = session.createProducer(dest);
					producer.setDeliveryMode(DeliveryMode.PERSISTENT);

					TextMessage msg = session.createTextMessage(msgStr);
					producer.setTimeToLive(TOPIC_TIME_TO_LIVE);
					producer.send(msg);
					// session.close();

					MessageDO msgDO = new MessageDO();
					msgDO.setId("mq_" + msg.getJMSMessageID());
					msgDO.setName(name);
					msgDO.setContent(msgStr);
					msgDO.setType("QUEUE");
					msgDO.setTid(tid);
					if (!messageService.create(msgDO)) {
						LOG.error("[save_topic_msg_fail] " + msgDO);
					}

					session.commit();
					session.close();

				} catch (JMSException e) {
					LOG.error("SEND_MQ_ERROR", e);
				}
			}
		});

		return true;
	}

	@Override
	public boolean listenTopic(String  processId, String topic, final ListenCallBack listenCallBack) {
	 
		try {
			if(CONSUMMER_MAP.containsKey(processId)){
				return true;
			}
			Session  session = CONNECTION.createSession(false, Session.AUTO_ACKNOWLEDGE); 
			MessageConsumer messageConsumer = session.createConsumer(new ActiveMQTopic(topic));
			messageConsumer.setMessageListener(new MessageListener() {

				@Override
				public void onMessage(Message message) {
					if (!(message instanceof TextMessage)) {
						return;
					}
					listenCallBack.receive(message);
				}
			});
			CONSUMMER_MAP.put(processId, session);
			return true;
		} catch (JMSException e) {
			e.printStackTrace();
		} finally {

		}
		return false;
	}

	@Override
	public boolean stopListen(String processId) {
		try{

			Session session = CONSUMMER_MAP.remove(processId);
			if(session==null){
				return true;
			}
			session.close();
			return true;
		}catch(Exception e){
			e.printStackTrace();
		}finally {
			
		}
		return false;
	}

	@Override
	public boolean listenQueue(String processId, String queue,final ListenCallBack listenCallBack) {
		try {
			if(CONSUMMER_MAP.containsKey(processId)){
				return true;
			}
			Session  session = CONNECTION.createSession(false, Session.AUTO_ACKNOWLEDGE); 
			MessageConsumer messageConsumer = session.createConsumer(new ActiveMQQueue(queue));
			messageConsumer.setMessageListener(new MessageListener() {

				@Override
				public void onMessage(Message message) {
					if (!(message instanceof TextMessage)) {
						return;
					}
					listenCallBack.receive(message);
				}
			});
			CONSUMMER_MAP.put(processId, session);
			return true;
		} catch (JMSException e) {
			e.printStackTrace();
		} finally {

		}
		return false;
	}

	@Override
	public Map<String, Session> getAllSession() {
		return CONSUMMER_MAP;
	}
	
}
